package com.lichao.process_function;

import com.lichao.source.Event;
import com.lichao.source_function.ClickSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.sources.wmstrategies.WatermarkStrategy;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

// ProcessFunction 数据处理
public class ProcessFunctionExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource());

//                .assignTimestampsAndWatermarks(
//                WatermarkStrategy.<Event>forMonotonousTimestamps()
//                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
//                            @Override
//                            public long extractTimestamp(Event event, long l) {
//                                return event.timestamp;
//                            }
//                        })
//        );

//        List<Event> userName = new ArrayList<>();
//        // MySQL CDC Source
//        MySqlSource<Row> mySqlSource = MySqlSource.<Row>builder()
//                .hostname("localhost")
//                .port(3306)
//                .databaseList("your_database")
//                .tableList("your_database.your_table")
//                .username("root")
//                .password("123456")
//                .deserializer(new JsonDebeziumDeserializationSchema()) // 根据你的需要选择合适的反序列化器
//                .build();
//
//        DataStream<Row> stream = env.addSource(mySqlSource);

        SingleOutputStreamOperator<String> process = stream.process(new ProcessFunction<Event, String>() {
            @Override
            public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
                out.collect(value.user);

//                userName.forEach(event -> {
//                    if (event.user.equals(value.user)){
//                        out.collect(value.user);
//                    }
//                });
                System.out.println("timestamp---" + ctx.timestamp());
                System.out.println(ctx.timerService().currentWatermark());
            }
        });

        process.print();
        env.execute();
    }
}